package rxgo
import(
	"context"
	"reflect"
	"sync"
	"time"
	"fmt"
)

type filteringOperator struct{
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool)
}

func(ftop filteringOperator) op(ctx context.Context, o *Observable){
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	go func() {
		end := false
		for x := range in {
			if end {
				continue
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error
			if e, ok := x.(error); ok && !o.flip_accept_error {
				o.sendToFlow(ctx, e, out)
				continue
			}
			// scheduler
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if ftop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				go func() {
					defer wg.Done()
					if ftop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}

		if o.flip != nil{
			buffer := reflect.ValueOf(o.flip)
			if buffer.Kind() != reflect.Slice {
				panic("buffer is not a slice")
			}
			for i := 0; i < buffer.Len(); i++ {
				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}
		wg.Wait() //waiting all go-routines completed
		o.closeFlow(out)
	}()
}

func (parent *Observable) Debounce(timespan time.Duration) (o *Observable){
	o = parent.newFilteringObservable("Debounce")

	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	o.threading = ThreadingComputing
	count := 0
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool){
			count++
			var temp = count
			time.Sleep(timespan)
			//time.Sleep(5*time.Microsecond)
			if temp == count{
				end = o.sendToFlow(ctx, item.Interface(), out)
			}
			return
	}}
	return o
}

func (parent *Observable) Distinct()(o *Observable){
	o = parent.newFilteringObservable("Distinct")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	m := map[string]bool{}
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			str := fmt.Sprintf("%v", item)
			if _, ok := m[str]; !ok{
				m[str] = true
				end = o.sendToFlow(ctx, item.Interface(), out)
			}
			return 
	}}
	return o
}

func (parent *Observable) ElementAt(n int)(o *Observable)  {
	o = parent.newFilteringObservable("ElementAt")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	count := 0
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			if count == n {
				end = o.sendToFlow(ctx, item.Interface(), out)
			}
			count++
			return
	}}
	return o
}

func (parent *Observable) First()(o *Observable){
	o = parent.newFilteringObservable("First")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	count := 0
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			if count == 0{
				o.sendToFlow(ctx, item.Interface(), out)
				end = true
			}
			count++
			return
	}}
	return o
}

func (parent *Observable) IgnoreElements()(o *Observable){
	o = parent.newFilteringObservable("IgnoreElements")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			return
	}}
	return o
}

func (parent *Observable) Last()(o *Observable){
	o = parent.newFilteringObservable("Last")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	//temp := make([]reflect.Value, 1)
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			//temp[0] = item
			o.flip = append([]interface{}{},item.Interface())
			return false
	}}
	return o
}

func (parent *Observable) Sample(timespan time.Duration)(o *Observable){
	o = parent.newTransformObservable("Sample")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	o.threading = ThreadingComputing
	temp := make([]reflect.Value, 128)
	flag := false
	count := 0
	var wg sync.WaitGroup
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool){
			temp[count] = item
			count = count + 1
			//fmt.Printf("append: %d\n", item.Interface().(int))
			if !flag{
				flag = true
				wg.Add(1)
				go func() {
					defer wg.Done()
					for {
						time.Sleep(timespan + 50*time.Millisecond)
						select{
							case <-ctx.Done():
								return
							default:
								if count == 0 {
									return
								}
								//fmt.Printf("sample %d with beforeitem %d\n", temp[count-1].Interface(), count-1)
								if o.sendToFlow(ctx, temp[count-1].Interface(), out){
									return
								}
								count = 0
						}
					}
				}()
				wg.Wait()
			} 
			return false
	}}
	return o
}

func (parent *Observable) Skip(n int)(o *Observable){
	o = parent.newFilteringObservable("Skip")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	count := 0
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			if count >= n{
				end = o.sendToFlow(ctx, item.Interface(), out)
			}
			count++
			return
	}}
	return o
}

func (parent *Observable) SkipLast(n int)(o *Observable){
	o = parent.newFilteringObservable("SkipLast")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	var temp []reflect.Value
	count := 0
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			if count == n{
				end = o.sendToFlow(ctx, temp[0].Interface(),out)
				temp = temp[1:]
			}else{
				count++
			}
			temp = append(temp, item)
			return
	}}
	return o
}

func (parent *Observable) Take (n int)(o *Observable){
	o = parent.newFilteringObservable("Take")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	count := 0
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			if count < n{
				end = o.sendToFlow(ctx, item.Interface(), out)
			}
			count++
			return
	}}
	return o
}

func (parent *Observable) TakeLast (n int)(o *Observable){
	o = parent.newFilteringObservable("TakeLast")
	o.flip_accept_error = false
	o.flip_sup_ctx = false
	o.flip = nil
	count := 0
	var temp []interface{}
	o.operator = filteringOperator{
		func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{})(end bool){
			count++
			if count <= n {
				temp = append(temp, item.Interface())
				//temp = temp[1:]
			}else{
				temp = temp[1:]
				temp = append(temp,item.Interface())
			}
			//temp = append(temp, item)

			o.flip = temp
			return
	}}
	return o
}

func (parent *Observable) newFilteringObservable(name string)(o *Observable){
	o = newObservable()
	o.Name = name
	parent.next = o
	o.pred = parent
	o.root = parent.root
	o.buf_len = BufferLen
	return o
}

